package main

import (
	"sync"
	"github.com/streadway/amqp"
	"time"
	"fmt"
	"log"
)

// Receiver 观察者模式需要的接口，观察者用于接收指定的queue到来的数据
type Receiver interface {
	QueueName() string     // 获取接收者需要监听的队列
	RouterKey() string     // 这个队列绑定的路由
	OnError(error)         // 处理遇到的错误，当RabbitMQ对象发生了错误，他需要告诉接收者处理错误
	OnReceive([]byte) bool // 处理收到的消息, 这里需要告知RabbitMQ对象消息是否处理成功
}

// RabbitMQ 用于管理和维护rabbitmq的对象
type RabbitMQ struct {
	wg sync.WaitGroup

	channel              *amqp.Channel
	exchangeName         string // exchange的名称
	exchangeType         string // exchange的类型
	receivers            []Receiver
	useFailRetryAndQueue bool

	failQueue *ConsumeFailQueue
}

type ConsumeFailQueue struct {
	MaxRetry   int    //重试最大次数
	RetryTime  int64  //每隔多少秒重试
	Exchange   string //失败 exchange
	RoutingKey string //失败 routekey
}

// 创建一个新的操作RabbitMQ的对象
func NewMQ(exchangeName, exchangeType string, useFailRetryAndQueue bool) *RabbitMQ {
	// 这里可以根据自己的需要去定义
	return &RabbitMQ{
		exchangeName:         exchangeName,
		exchangeType:         exchangeType,
		useFailRetryAndQueue: useFailRetryAndQueue,
	}
}

// 准备rabbitmq的Exchange
func (mq *RabbitMQ) prepareExchange() error {
	// 声明 Exchange
	err := mq.channel.ExchangeDeclare(
		mq.exchangeName, // exchange
		mq.exchangeType, // type
		true,            // durable
		false,           // autoDelete
		false,           // internal
		false,           // noWait
		nil,             // args
	)

	if nil != err {
		return err
	}

	return nil
}

// Start 启动Rabbitmq的客户端
func (mq *RabbitMQ) Start() {
	url := "amqp://admin:123456@127.0.0.1:5672/"
	conn, err := amqp.Dial(url)
	defer conn.Close()
	if err != nil {
		log.Printf("%s: %s", "Failed to connect to RabbitMQ", err)
		return
	}
	// 获取新的channel对象
	ch, err := conn.Channel()
	defer ch.Close()
	if err != nil {
		log.Printf("%s: %s", "Failed to open a channel", err)
		return
	}
	mq.channel = ch
	// 初始化Exchange
	mq.prepareExchange()

	for _, receiver := range mq.receivers {
		mq.wg.Add(1)
		// 每个接收者单独启动一个goroutine用来初始化queue并接收消息
		go mq.listen(receiver)
	}

	mq.wg.Wait()
}

// RegisterReceiver 注册一个用于接收指定队列指定路由的数据接收者
func (mq *RabbitMQ) RegisterReceiver(receiver Receiver) {
	mq.receivers = append(mq.receivers, receiver)
	log.Println(mq.receivers)
}

// Listen 监听指定路由发来的消息
// 这里需要针对每一个接收者启动一个goroutine来执行listen
// 该方法负责从每一个接收者监听的队列中获取数据，并负责重试
func (mq *RabbitMQ) listen(receiver Receiver) {
	defer mq.wg.Done()

	// 这里获取每个接收者需要监听的队列和路由
	queueName := receiver.QueueName()
	routerKey := receiver.RouterKey()
	log.Println(queueName, ":", routerKey)
	// 申明Queue
	_, err := mq.channel.QueueDeclare(
		queueName, // name
		true,      // durable
		false,     // delete when usused
		false,     // exclusive(排他性队列)
		false,     // no-wait
		nil,       // arguments
	)
	if nil != err {
		// 当队列初始化失败的时候，需要告诉这个接收者相应的错误
		receiver.OnError(fmt.Errorf("初始化队列 %s 失败: %s", queueName, err.Error()))
	}

	// 将Queue绑定到Exchange上去
	err = mq.channel.QueueBind(
		queueName,       // queue name
		routerKey,       // routing key
		mq.exchangeName, // exchange
		false,           // no-wait
		nil,
	)
	if nil != err {
		receiver.OnError(fmt.Errorf("绑定队列 [%s - %s] 到交换机失败: %s", queueName, routerKey, err.Error()))
	}

	// 获取消费通道 确保rabbitmq会一个一个发消息
	mq.channel.Qos(1, 0, true)
	msgs, err := mq.channel.Consume(
		queueName, // queue
		"",        // consumer
		false,     // auto-ack
		false,     // exclusive
		false,     // no-local
		false,     // no-wait
		nil,       // args
	)
	if nil != err {
		receiver.OnError(fmt.Errorf("获取队列 %s 的消费通道失败: %s", queueName, err.Error()))
	}

	// 使用callback消费数据
	for msg := range msgs {
		// 当接收者消息处理失败的时候，
		// 比如网络问题导致的数据库连接失败，redis连接失败等等这种
		// 通过重试可以成功的操作，那么这个时候是需要重试的
		// 直到数据处理成功后再返回，然后才会回复rabbitmq ack
		times := 0
		for !receiver.OnReceive(msg.Body) {

			if mq.useFailRetryAndQueue {

				if times > mq.failQueue.MaxRetry {
					//超出重试次数 退出
					err = mq.channel.Publish(
						mq.failQueue.Exchange,   // exchange
						mq.failQueue.RoutingKey, // routing key
						false,                   // mandatory
						false,                   // immediate
						amqp.Publishing{
							ContentType: "text/plain",
							Body:        []byte(msg.Body),
						})

					break
				}
				//do
				log.Println("receiver 数据处理失败，将要重试")
				time.Sleep(time.Duration(mq.failQueue.RetryTime) * time.Second)
				times++

			} else {
				log.Println("不处理失败消息：", string(msg.Body))
				break
			}

		}
		// 确认收到本条消息, multiple必须为false
		msg.Ack(false)

	}
}

//主队列接受
type RootReceiver struct {
	Queue  string
	Router string
}

func (a RootReceiver) QueueName() string {
	return a.Queue
}
func (a RootReceiver) RouterKey() string {
	return a.Router
}
func (a RootReceiver) OnError(err error) {
	log.Println(err)
}
func (a RootReceiver) OnReceive(bt []byte) bool {
	log.Println(a.QueueName() ,":", string(bt))
	//测试 内容为0 是消费失败
	if string(bt) == "0" {
		return false
	}
	return true
}

//失败队列接受
type FailReceiver struct {
	Queue  string
	Router string
}

func (f FailReceiver) QueueName() string {
	return f.Queue
}
func (f FailReceiver) RouterKey() string {
	return f.Router
}
func (f FailReceiver) OnError(err error) {
	log.Println(err)
}
func (f FailReceiver) OnReceive(bt []byte) bool {
	log.Println(f.QueueName() ,":", string(bt))
	if string(bt) == "0" {
		return false
	}
	return true
}

// 启动并开始处理数据
func main() {

	// 主消费者
	r := RootReceiver{"test.queue", "test.key"}
	r2 := RootReceiver{"test2.queue", "test.key"}
	mq := NewMQ("test.direct", "direct", true)
	mq.failQueue = &ConsumeFailQueue{
		MaxRetry:   10,                 //重试最大次数 重试失败消息会进入失败队列
		RetryTime:  1,                  //每隔多少秒重试
		Exchange:   "test.fail.direct", //失败 exchange
		RoutingKey: "test.fail.key",    //失败 routekey
	}
	// 将这个接收者注册到
	mq.RegisterReceiver(r)
	mq.RegisterReceiver(r2)
	go mq.Start()

	// 失败消费者
	d := FailReceiver{"test.fail.queue", "test.fail.key"}
	dmq := NewMQ("test.fail.direct", "direct", false)
	// 将这个接收者注册到
	dmq.RegisterReceiver(d)
	go dmq.Start()

	select {}

}
